package com.zhida;

import io.reactivex.Emitter;
import io.reactivex.Flowable;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import okhttp3.OkHttpClient;
import okhttp3.ResponseBody;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import retrofit2.Retrofit;
import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory;
import retrofit2.converter.scalars.ScalarsConverterFactory;
import rx.functions.Action1;
import rx.schedulers.Schedulers;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Administrator on 2017-04-05.
 */
public class SpiderTest {

    private final static Logger logger = LoggerFactory.getLogger(SpiderTest.class);

    private final static String BASE_URL = "http://fec.mofcom.gov.cn";

    private BlockingQueue<String> queue = new LinkedBlockingQueue<String>();

    private ListProcessor listProcessor = new ListProcessor();

    private OkHttpClient.Builder httpClient = new OkHttpClient.Builder();

    private Retrofit retrofit = new Retrofit.Builder()
            .baseUrl(BASE_URL)
            .addConverterFactory(ScalarsConverterFactory.create())//转换字符串
//                .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
            .build();

    private final AtomicInteger runningCount = new AtomicInteger(0);

    public void start() {

        Disposable stats = Flowable.interval(1, 2, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {
                    public void accept(@NonNull Long aLong) throws Exception {
                        logger.info("crawler concurrency {}, {} requests are waiting scheduled"
                                , runningCount.get(), queue.size());
                    }
                });

        queue.offer("qwe");

        Flowable.generate(new Consumer<Emitter<String>>() {
            public void accept(@NonNull Emitter<String> stringEmitter) throws Exception {
                String poll = queue.poll();
                if (poll != null) {
                    stringEmitter.onNext(poll);
                }
            }
        })
                .subscribeOn(io.reactivex.schedulers.Schedulers.from(Executors.newSingleThreadExecutor()))
                .doOnNext(new Consumer<String>() {
                    public void accept(@NonNull String s) throws Exception {
                        runningCount.incrementAndGet();
                    }
                })
                .flatMap(new Function<String, Publisher<?>>() {
                    public Publisher<?> apply(@NonNull String s) throws Exception {
                        queue.offer("qwe");
                        return Flowable.just(11);
                    }
                }, 5)
                .doOnNext(new Consumer<Object>() {
                    public void accept(@NonNull Object o) throws Exception {
                        runningCount.decrementAndGet();
                    }
                })
                .blockingSubscribe();

        /*
        Observable.create(new Observable.OnSubscribe<String>() {
            public void call(Subscriber<? super String> subscriber) {
                String poll = queue.poll();
                subscriber.onNext(poll);
                subscriber.onCompleted();
            }
        })
                .doOnNext(new Action1<String>() {
                    public void call(String s) {
                        runningCount.incrementAndGet();
                    }
                })
                .flatMap(new Func1<String, Observable<?>>() {
                    public Observable<?> call(String s) {
                        queue.offer(s);
                        return Observable.just(1, 2, 3);
                    }
                }, 5)
                .doOnNext(new Action1<Object>() {
                    public void call(Object o) {
                        runningCount.decrementAndGet();
                    }
                })
                .toBlocking();
        */

                /*.subscribe(new Subscriber<String>() {
                    public void onCompleted() {
                        System.out.println("onCompleted");
                    }

                    public void onError(Throwable e) {

                    }

                    public void onNext(String s) {
                        System.out.println(s);
                    }
                });*/
    }

}
